package com.kingstar.ceres.runable;

import com.baomidou.mybatisplus.core.toolkit.SystemClock;
import com.kingstar.ceres.entity.secondary.*;
import com.kingstar.ceres.memory.MessageMemory;
import com.kingstar.ceres.service.secondary.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
@Order(Integer.MIN_VALUE)
@Slf4j
public class SaveToDbRunnable implements ApplicationRunner {

    @Autowired
    KCmdsIrsMarketDealService irsMarketDealService;

    @Autowired
    KCmdsBondMarketDealService bondMarketDealService;

    @Autowired
    KCmdsMarketDataSummaryBondService marketDataSummaryBondService;

    @Autowired
    KCmdsXbondMarketDealService xbondMarketDealService;

    @Autowired
    KCmdsXswapMarketDealService xswapMarketDealService;

    @Value("${com.kingstar.ceres.save.pageSize}")
    public Integer pageSize;

    @Value("${com.kingstar.ceres.save.waitTime}")
    public Long waitTime;


    @Override
    public void run(ApplicationArguments args) throws Exception {

        Thread irs = new Thread(() -> {
            List<KCmdsIrsMarketDeal> result = new ArrayList();
            long begin = SystemClock.now();
            while (true){
                KCmdsIrsMarketDeal poll = MessageMemory.IRS_MARKET_DEAL_QUEUE.poll();
                if (poll!=null){
                    result.add(poll);
                }
                if (result.size()==pageSize || (result.size()>0 &&(SystemClock.now()-begin)>waitTime) ){
                    try {
                        boolean b = irsMarketDealService.saveBatch(result);
                        if (b){
                            log.info("save to KCmdsIrsMarketDeal:{}",result.size());
                            result.clear();
                            begin=SystemClock.now();
                        }else {
                            log.error("==============kCmdsIrsMarketDealService saveBatch err1==================");
                        }
                    } catch (Exception e) {
                        log.error("==============kCmdsIrsMarketDealService saveBatch err2==================");
                        e.printStackTrace();
                    }
                }
            }
        });
        irs.start();

        Thread bond = new Thread(() -> {
            List<KCmdsBondMarketDeal> result = new ArrayList();
            long begin = SystemClock.now();
            while (true){
                KCmdsBondMarketDeal poll = MessageMemory.BOND_MARKET_DEAL_QUEUE.poll();
                if (poll!=null){
                    result.add(poll);
                }
                if (result.size()==pageSize || (result.size()>0 &&(SystemClock.now()-begin)>waitTime) ){
                    try {
                        boolean b = bondMarketDealService.saveBatch(result);
                        if (b){
                            log.info("save to KCmdsBondMarketDeal:{}",result.size());
                            result.clear();
                            begin=SystemClock.now();
                        }else {
                            log.error("==============bondMarketDealService saveBatch err1==================");
                        }
                    } catch (Exception e) {
                        log.error("==============bondMarketDealService saveBatch err2==================");
                        e.printStackTrace();
                    }
                }
            }
        });
        bond.start();


        Thread summary = new Thread(() -> {
            List<KCmdsMarketDataSummaryBond> result = new ArrayList();
            long begin = SystemClock.now();
            while (true){
                KCmdsMarketDataSummaryBond poll = MessageMemory.MARKET_DATA_SUMMARY_BOND_QUEUE.poll();
                if (poll!=null){
                    result.add(poll);
                }
                if (result.size()==pageSize || (result.size()>0 &&(SystemClock.now()-begin)>waitTime) ){
                    try {
                        boolean b = marketDataSummaryBondService.saveBatch(result);
                        if (b){
                            log.info("save to KCmdsMarketDataSummaryBond:{}",result.size());
                            result.clear();
                            begin=SystemClock.now();
                        }else {
                            log.error("==============marketDataSummaryBondService saveBatch err1==================");
                        }
                    } catch (Exception e) {
                        log.error("==============marketDataSummaryBondService saveBatch err2==================");
                        e.printStackTrace();
                    }
                }
            }
        });
        summary.start();

        Thread xBond = new Thread(() -> {
            List<KCmdsXbondMarketDeal> result = new ArrayList();
            long begin = SystemClock.now();
            while (true){
                KCmdsXbondMarketDeal poll = MessageMemory.XBOND_MARKET_DEAL_QUEUE.poll();
                if (poll!=null){
                    result.add(poll);
                }
                if (result.size()==pageSize || (result.size()>0 &&(SystemClock.now()-begin)>waitTime) ){
                    try {
                        boolean b = xbondMarketDealService.saveBatch(result);
                        if (b){
                            log.info("save to KCmdsXbondMarketDeal:{}",result.size());
                            result.clear();
                            begin=SystemClock.now();
                        }else {
                            log.error("==============xbondMarketDealService saveBatch err1==================");
                        }
                    } catch (Exception e) {
                        log.error("==============xbondMarketDealService saveBatch err2==================");
                        e.printStackTrace();
                    }
                }
            }
        });
        xBond.start();

        Thread xSwap = new Thread(() -> {
            List<KCmdsXswapMarketDeal> result = new ArrayList();
            long begin = SystemClock.now();
            while (true){
                KCmdsXswapMarketDeal poll = MessageMemory.XSWAP_MARKET_DEAL_QUEUE.poll();
                if (poll!=null){
                    result.add(poll);
                }
                if (result.size()==pageSize || (result.size()>0 &&(SystemClock.now()-begin)>waitTime) ){
                    try {
                        boolean b = xswapMarketDealService.saveBatch(result);
                        if (b){
                            log.info("save to KCmdsXswapMarketDeal:{}",result.size());
                            result.clear();
                            begin=SystemClock.now();
                        }else {
                            log.error("==============xswapMarketDealService saveBatch err1==================");
                        }
                    } catch (Exception e) {
                        log.error("==============xswapMarketDealService saveBatch err2==================");
                        e.printStackTrace();
                    }
                }
            }
        });
        xSwap.start();

    }
}
